package main

import (
	"encoding/csv"
	"fmt"
	"log"
	"os"
	"strconv"
	"strings"

	"github.com/zeromicro/go-zero/core/mr"
)

// AppRecord 表示原始应用记录
type AppRecord struct {
	ID                string
	TrackName         string
	SizeBytes         int64
	Currency          string
	Price             float64
	RatingCountTotal  int
	RatingCountVer    int
	UserRating        float64
	UserRatingVer     float64
	Version           string
	ContentRating     string
	PrimeGenre        string
	SupportedDevices  int
	IpadScreenShots   int
	Languages         int
	VPP_LICENSE       int
}

// CleanedAppRecord 表示清洗后的应用记录
type CleanedAppRecord struct {
	ID               string
	Name             string
	SizeMB           float64 // 转换为MB单位
	Price            float64
	Genre            string
	UserRating       float64
	ContentRating    string
	RatingCountTotal int
}

func main() {
	// 读取CSV文件
	records, err := readCSVFile("AppleStore.csv")
	if err != nil {
		log.Fatal("读取CSV文件出错:", err)
	}

	fmt.Printf("总共读取到 %d 条记录\n", len(records))

	// 使用MapReduce清洗数据
	cleanedRecords, err := cleanAppData(records)
	if err != nil {
		log.Fatal("数据清洗过程中出现错误:", err)
	}

	fmt.Printf("清洗后剩余 %d 条有效记录\n", len(cleanedRecords))

	// 显示前10条清洗后的数据
	// fmt.Println("\n前10条清洗后的数据:")
	// for i, record := range cleanedRecords {
	// 	if i >= 10 {
	// 		break
	// 	}
	// 	fmt.Printf("ID: %s, 名称: %s, 大小: %.2f MB, 价格: $%.2f, 类型: %s, 评分: %.1f, 内容评级: %s, 评分总数: %d\n",
	// 		record.ID, record.Name, record.SizeMB, record.Price, record.Genre, record.UserRating, record.ContentRating, record.RatingCountTotal)
	// }

	// 将清洗后的数据写入新的CSV文件
	err = writeCleanedDataToCSV("cleaned_applestore.csv", cleanedRecords)
	if err != nil {
		log.Fatal("写入清洗后数据出错:", err)
	}
	fmt.Println("\n清洗后的数据已保存至 cleaned_applestore.csv 文件")

	// 统计各类型应用数量
	genreStats := countByGenre(cleanedRecords)
	fmt.Println("\n各类型应用数量统计:")
	for genre, count := range genreStats {
		fmt.Printf("%s: %d\n", genre, count)
	}
}

// readCSVFile 读取CSV文件
func readCSVFile(filename string) ([][]string, error) {
	file, err := os.Open(filename)
	if err != nil {
		return nil, err
	}
	defer file.Close()

	reader := csv.NewReader(file)
	records, err := reader.ReadAll()
	if err != nil {
		return nil, err
	}

	// 返回除了标题行之外的所有记录
	return records[1:], nil
}

// writeCleanedDataToCSV 将清洗后的数据写入CSV文件
func writeCleanedDataToCSV(filename string, records []CleanedAppRecord) error {
	file, err := os.Create(filename)
	if err != nil {
		return err
	}
	defer file.Close()

	writer := csv.NewWriter(file)
	defer writer.Flush()

	// 写入标题行
	headers := []string{"ID", "Name", "SizeMB", "Price", "Genre", "UserRating", "ContentRating", "RatingCountTotal"}
	err = writer.Write(headers)
	if err != nil {
		return err
	}

	// 写入数据行
	for _, record := range records {
		row := []string{
			record.ID,
			record.Name,
			fmt.Sprintf("%.2f", record.SizeMB),
			fmt.Sprintf("%.2f", record.Price),
			record.Genre,
			fmt.Sprintf("%.1f", record.UserRating),
			record.ContentRating,
			strconv.Itoa(record.RatingCountTotal),
		}
		err = writer.Write(row)
		if err != nil {
			return err
		}
	}

	return nil
}

// cleanAppData 使用MapReduce清洗应用数据
func cleanAppData(data [][]string) ([]CleanedAppRecord, error) {
	var result []CleanedAppRecord

	// Generate function - 生成数据源
	generateFunc := func(source chan<- []string) {
		for _, record := range data {
			source <- record
		}
	}

	// Map function - 执行过滤和转换逻辑
	mapperFunc := func(record []string, writer mr.Writer[CleanedAppRecord], cancel func(error)) {
		// 阶段一: 过滤逻辑
		if len(record) < 16 {
			// 记录不完整，跳过
			return
		}

		// 阶段二: 转换逻辑
		cleanedRecord, err := transformAppRecord(record)
		if err != nil {
			// 转换失败，跳过该记录
			return
		}

		writer.Write(cleanedRecord)
	}

	// Reduce function - 收集清洗后的数据
	reducerFunc := func(pipe <-chan CleanedAppRecord, writer mr.Writer[[]CleanedAppRecord], cancel func(error)) {
		var cleanedRecords []CleanedAppRecord
		for record := range pipe {
			cleanedRecords = append(cleanedRecords, record)
		}
		writer.Write(cleanedRecords)
	}

	// 执行MapReduce操作
	result, err := mr.MapReduce(generateFunc, mapperFunc, reducerFunc)
	if err != nil {
		return nil, err
	}

	return result, nil
}

// transformAppRecord 处理单条应用记录的转换逻辑
func transformAppRecord(record []string) (CleanedAppRecord, error) {
	// 解析ID
	id := strings.TrimSpace(record[0])
	if id == "" {
		return CleanedAppRecord{}, fmt.Errorf("ID不能为空")
	}

	// 解析名称
	name := strings.TrimSpace(record[1])
	if name == "" {
		return CleanedAppRecord{}, fmt.Errorf("应用名称不能为空")
	}

	// 解析大小并转换为MB
	sizeBytesStr := strings.TrimSpace(record[2])
	sizeBytes, err := strconv.ParseInt(sizeBytesStr, 10, 64)
	if err != nil || sizeBytes <= 0 {
		return CleanedAppRecord{}, fmt.Errorf("无效的应用大小: %s", sizeBytesStr)
	}
	sizeMB := float64(sizeBytes) / (1024 * 1024) // 转换为MB

	// 解析价格
	priceStr := strings.TrimSpace(record[4])
	price, err := strconv.ParseFloat(priceStr, 64)
	if err != nil {
		return CleanedAppRecord{}, fmt.Errorf("无效的价格: %s", priceStr)
	}

	// 解析用户评分
	userRatingStr := strings.TrimSpace(record[7])
	userRating, err := strconv.ParseFloat(userRatingStr, 64)
	if err != nil || userRating < 0 || userRating > 5 {
		return CleanedAppRecord{}, fmt.Errorf("无效的用户评分: %s", userRatingStr)
	}

	// 解析内容评级
	contentRating := strings.TrimSpace(record[10])
	if contentRating == "" {
		return CleanedAppRecord{}, fmt.Errorf("内容评级不能为空")
	}

	// 解析类型
	genre := strings.TrimSpace(record[11])
	if genre == "" {
		return CleanedAppRecord{}, fmt.Errorf("应用类型不能为空")
	}

	// 解析总评分数
	ratingCountTotalStr := strings.TrimSpace(record[5])
	ratingCountTotal, err := strconv.Atoi(ratingCountTotalStr)
	if err != nil || ratingCountTotal < 0 {
		return CleanedAppRecord{}, fmt.Errorf("无效的总评分数: %s", ratingCountTotalStr)
	}

	return CleanedAppRecord{
		ID:               id,
		Name:             name,
		SizeMB:           sizeMB,
		Price:            price,
		Genre:            genre,
		UserRating:       userRating,
		ContentRating:    contentRating,
		RatingCountTotal: ratingCountTotal,
	}, nil
}

// countByGenre 统计各类型应用数量
func countByGenre(records []CleanedAppRecord) map[string]int {
	result := make(map[string]int)

	// 使用mr.MapReduce进行统计
	generateFunc := func(source chan<- CleanedAppRecord) {
		for _, record := range records {
			source <- record
		}
	}

	mapperFunc := func(record CleanedAppRecord, writer mr.Writer[map[string]int], cancel func(error)) {
		count := make(map[string]int)
		count[record.Genre] = 1
		writer.Write(count)
	}

	reducerFunc := func(pipe <-chan map[string]int, writer mr.Writer[map[string]int], cancel func(error)) {
		result := make(map[string]int)
		for count := range pipe {
			for genre, cnt := range count {
				result[genre] += cnt
			}
		}
		writer.Write(result)
	}

	result, err := mr.MapReduce(generateFunc, mapperFunc, reducerFunc)
	if err != nil {
		// 出错时使用传统方法
		result = make(map[string]int)
		for _, record := range records {
			result[record.Genre]++
		}
	}

	return result
}